package com.atguigu.tabletest

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.descriptors.{Csv, Kafka, Schema}


/**
 *
 * @description: dataStream转成table 输出到外部系统
 * @time: 2021-03-15 15:35
 * @author: baojinlong
 **/
object Example04KafkaPipeline {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)

    // 创建表的执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment)
    // 连接到kafka
    tableEnv.connect(
      new Kafka()
        .version("0.11")
        .topic("sensor")
        .property("bootstrap.servers", "localhost:9092")
        .property("zookeeper.connect", "localhost:2182")
    )
      .withFormat(new Csv)
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING)
          .field("timestamp", DataTypes.BIGINT)
          .field("temperature", DataTypes.DOUBLE)
      )
      .createTemporaryTable("kafkaInputTable")

    // 查询转换
    // 进行转换操作
    val sensorTable: Table = tableEnv.from("kafkaInputTable")
    val resultTable: Table = sensorTable
      .select('id, 'temperature)
      .filter('id === "sensor_1")
    // 聚合转换
    val aggTable: Table = sensorTable
      .groupBy("id") // 基于id分组
      .select('id, 'id.count as 'count)


    // 输出到kafka
    tableEnv.connect(
      new Kafka()
        .version("0.11")
        .topic("sensor-test")
        .property("bootstrap.servers", "localhost:9092")
        .property("zookeeper.connect", "localhost:2182")
    )
      .withFormat(new Csv)
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING)
          .field("temperature", DataTypes.DOUBLE)
      )
      .createTemporaryTable("kafkaOutputTable")
    resultTable.insertInto("kafkaOutputTable")
    // aggTable底层还是appendStream,所以还是不能直接写入到kafka中 sql本身适用于批处理 流处理有很大差别,动态表
    environment.execute("kafka pipeline test")
  }

}
